package cn.stalk.compare;

import cn.stalk.compare.result.ComparisonResult;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

/**
 * 流式数据比较器
 * <pre>
 * 1.适用于大数据量对比
 * 2.对比的双方数据都必须按业务主键顺序排列【非常重要】
 * 3.业务主键的排序顺序默认是nullsLast，但也可以设置，需要保证设置与传递的数据一致
 * 4.源数据与目标数据都需要在多线程环境下加载，主线程通过start开始比较
 * </pre>
 */
public class StreamComparator<T> {

    private Consumer<T> dataHandler = null;
    private final DataComparator<T> dataComparator;
    private final Consumer<ComparisonResult<T>> handler;

    private LinkedBlockingQueue<DataWrapper<T>> sourceQueue = null;
    private LinkedBlockingQueue<DataWrapper<T>> targetQueue = null;
    private AtomicBoolean sourceStoped = new AtomicBoolean(false);
    private AtomicBoolean targetStoped = new AtomicBoolean(false);

    public StreamComparator(DataComparator<T> dataComparator, Consumer<ComparisonResult<T>> handler, int capacity) {
        this.dataComparator = dataComparator;
        this.handler = handler;
        sourceQueue = new LinkedBlockingQueue<>(capacity);
        targetQueue = new LinkedBlockingQueue<>(capacity);
    }

    public StreamComparator(DataComparator<T> dataComparator, Consumer<ComparisonResult<T>> handler) {
        this(dataComparator, handler, 1000);
    }

    public void setDataHandler(Consumer<T> dataHandler) {
        this.dataHandler = dataHandler;
    }

    /**
     * 开始比较
     */
    public void start() {
        LinkedList<DataWrapper<T>> sourceWrapperList = new LinkedList<>();
        LinkedList<DataWrapper<T>> targetWrapperList = new LinkedList<>();
        List<DataWrapper<T>> empty = new ArrayList<>();

        // 先以source为主来对比
        for (; ; ) {
            loadData(sourceWrapperList, sourceQueue, sourceStoped);
            if (sourceWrapperList.isEmpty()) {
                // source已读取完
                break;
            }
            loadData(targetWrapperList, targetQueue, targetStoped);

            ComparisonResult<T> result = null;
            if (targetWrapperList.isEmpty()) {
                // target已读取完了，source多余
                result = dataComparator.doCompare(sourceWrapperList, empty);
                sourceWrapperList.clear();
            } else {
                int rs = sourceWrapperList.getLast().idCompare(targetWrapperList.getLast());
                if (rs < 0) {
                    // source数据比较小，target缺数据
                    result = dataComparator.doCompare(sourceWrapperList, empty);
                    sourceWrapperList.clear();
                } else if (rs > 0) {
                    // source数据比较大，source缺数据
                    result = dataComparator.doCompare(empty, targetWrapperList);
                    targetWrapperList.clear();
                } else {
                    // id一样
                    result = dataComparator.doCompare(sourceWrapperList, targetWrapperList);
                    sourceWrapperList.clear();
                    targetWrapperList.clear();
                }
            }
            handler.accept(result);
        }

        // target数据都算多余
        for (; ; ) {
            loadData(targetWrapperList, targetQueue, targetStoped);
            if (targetWrapperList.isEmpty()) {
                break;
            }
            ComparisonResult<T> result = dataComparator.doCompare(empty, targetWrapperList);
            targetWrapperList.clear();
            handler.accept(result);
        }
    }

    /**
     * 从队列里取出数据放入wrapperList，需wrapperList无数据，或与wrapperList里数据的id一致才取出数据
     */
    private void loadData(LinkedList<DataWrapper<T>> wrapperList, LinkedBlockingQueue<DataWrapper<T>> queue, AtomicBoolean stoped) {
        DataWrapper<T> wrapper = null;
        for (; ; ) {
            wrapper = queue.peek();
            if (null == wrapper) {
                if (stoped.get()) {
                    // 停止时，再试图读取一次，仍为null时才退出
                    if (null == (wrapper = queue.peek())) {
                        break;
                    }
                } else {
                    continue;
                }
            }

            // 此时source一定有值
            if (wrapperList.isEmpty() || wrapperList.getLast().idCompare(wrapper) == 0) {
                queue.poll();
                wrapperList.add(wrapper);
                continue;
            }
            break;
        }
    }

    /**
     * 将源数据推送到队列中
     */
    public void pushSource(T source) {
        if (null != dataHandler) {
            dataHandler.accept(source);
        }
        try {
            sourceQueue.put(new DataWrapper<>(source, dataComparator.getKeyDataAccessorMap()));
        } catch (InterruptedException e) {
            stopSource();
            throw new RuntimeException("InterruptedException", e);
        }
    }

    /**
     * 将目标数据推送到队列中
     */
    public void pushTarget(T target) {
        if (null != dataHandler) {
            dataHandler.accept(target);
        }
        try {
            targetQueue.put(new DataWrapper<>(target, dataComparator.getKeyDataAccessorMap()));
        } catch (InterruptedException e) {
            stopTarget();
            throw new RuntimeException("InterruptedException", e);
        }
    }

    /**
     * 设置源数据已停止推送
     */
    public void stopSource() {
        sourceStoped.set(true);
    }

    /**
     * 设置目标数据已停止推送
     */
    public void stopTarget() {
        targetStoped.set(true);
    }

}
